package com.example.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author zhy
 * @version 1.0.0
 * @ClassName KafkaProducer.java
 * @Description TODO
 * @createTime 2021年07月03日 15:52:00
 */
@RestController
public class KafkaProducer {

    private static final String TOPIC_NAME = "topic1";

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    /**
     * 同步发送消息
     * @param nomalMessage
     */
    @RequestMapping(value = "/kafka/normal/{message}")
    @Transactional
    public void sendMessageSync(@PathVariable("message") String nomalMessage) {
        kafkaTemplate.send(TOPIC_NAME, nomalMessage);
    }

    /**
     * 异步发送消息
     * @param callbackMessage
     */
    @RequestMapping("/kafka/callbackOne/{message}")
    public void sendMessageAsync(@PathVariable("message") String callbackMessage) {
        kafkaTemplate.send(TOPIC_NAME, callbackMessage).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable ex) {
                System.out.println("发送消息失败：" + ex.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> result) {
                System.out.println("发送消息成功：" + result.getRecordMetadata().topic() + "-" +
                        result.getRecordMetadata().partition() + "-" + result.getRecordMetadata().offset());
            }
        });
    }

    /**
     *
     * @param keyMessage
     * @param key
     */
    @GetMapping("/kafka/key/{message}/{key}")
    public void sendMessage3(@PathVariable("message") String keyMessage, @PathVariable("key") String key) {
        kafkaTemplate.send(TOPIC_NAME, key, keyMessage);
    }

    @GetMapping("/kafka/transaction")
    public void sendMessage4() {
        kafkaTemplate.executeInTransaction(operations -> {
            operations.send(TOPIC_NAME, "test executeInTransaction");
            throw new RuntimeException("fail");
        });

        kafkaTemplate.send(TOPIC_NAME, "test executeInTransaction");
        throw new RuntimeException("fail");
    }

    @GetMapping("/kafka/partition/{message}")
    public void sendMessage5(@PathVariable("message") String message) {
//        监听topic1的0号分区，同时监听topic2的0号分区和topic2的1号分区里面offset从8开始的消息
        kafkaTemplate.send("topic3", 0, "sendMessage5", message);
        kafkaTemplate.send("topic2", 0, "sendMessage5", message);
        kafkaTemplate.send("topic2", 1, "sendMessage5", message);
    }
}
